package com.motu.vertx.module.utility.kafka;

import com.motu.vertx.module.utility.toolset.Tool;
import io.vertx.core.Vertx;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import io.vertx.kafka.client.producer.KafkaProducer;
import io.vertx.kafka.client.producer.KafkaProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class KafkaLogManager {

	public final static Logger logger = LoggerFactory.getLogger("KafkaLogManager");
	public static KafkaProducer<String, String> logProducer;// 日志kafka生产者

	public static void init(Vertx vertx, JsonObject appConfig) {

		String address = appConfig.getString("log-kafka-address");
		if (Tool.isEmpty(address)) {
			address = appConfig.getString("kafka-address");
			if (Tool.isEmpty(address)) {
				logger.error("KafkaLogManager init failed, address is empty");
				return;
			}
		}

		Properties log_config = new Properties();

		log_config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, address);
		log_config.put("key.serializer", StringSerializer.class);
		log_config.put("value.serializer", StringSerializer.class);
		log_config.put("acks", "1");

		// 优化kafka生产者配置
		KafkaManager.upgradeProducerConfig(log_config, appConfig);

		// 创建日志kafka生产者
		logProducer = KafkaProducer.create(vertx, log_config);
		logger.info("KafkaLogManager init success address:{}", address);
	}


	/**
	 * 添加游戏日志到kafka队列
	 */
	public static void addGameLog(String sql, JsonArray params, int dbType) {
		if (logProducer == null){
			return;
		}
		JsonObject obj = new JsonObject().put("sql", sql).put("params", params).put("dbtype", dbType);
		KafkaProducerRecord<String, String> record = KafkaProducerRecord.create("db_log", obj.toString());
		logProducer.send(record, done -> {
			if (done.failed()) {
				logger.error("addGameLog pushDBQueue fail cause = " + Tool.getException(done.cause()));
			}
		});
	}

	/**
	 * 添加消耗获得日志到kafka队列
	 */
	public static void addItemRecordLog(JsonArray params) {
		if (logProducer == null){
			return;
		}
		KafkaProducerRecord<String, String> record = KafkaProducerRecord.create("clickhouse_log", params.toString());
		logProducer.send(record, done -> {
			if (done.failed()) {
				logger.error("addItemRecordLog pushDBQueue fail cause = " + Tool.getException(done.cause()));
			}
		});
	}

	/**
	 * 添加协议监控日志到kafka队列
	 */
	public static void addProtocolLog(JsonArray params) {
		if (logProducer == null){
			return;
		}
		KafkaProducerRecord<String, String> record = KafkaProducerRecord.create("protocol_log", params.toString());
		logProducer.send(record, done -> {
			if (done.failed()) {
				logger.error("addProtocolLog pushDBQueue fail cause = " + Tool.getException(done.cause()));
			}
		});
	}

}
